package com.cike.sparkstudy.sparkstreaming.java;

import com.google.common.base.Optional;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.ArrayList;

/**
 * 基于trantionsform的实时广告计费日志黑名单过滤
 * 真实
 */
public class TransformBlacklist {
    @SuppressWarnings("deprecation")
    public static void main(String[] args){
        //创建实时计算对象
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("TransformBlacklist");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

        /**
         * 用户对我们的网站上的广告进行点击
         * 点击后进行实时计费，点一次算一次钱
         * 对于刷广告的商家，有一个黑名单
         * 只要进入黑名单中的用户点击广告都给过滤掉
         */
        //1、先配置一份黑名单
        ArrayList<Tuple2<String, Boolean>> blacklist = new ArrayList<Tuple2<String, Boolean>>();
        blacklist.add(new Tuple2<String, Boolean>("tom",true));
        final JavaPairRDD<String, Boolean> blacklistRDD = jssc.sc().parallelizePairs(blacklist);

        //2、这里是获取日志信息，格式为date username的形式
        JavaReceiverInputDStream<String> adsClickLogDStream = jssc.socketTextStream("bigdata", 9999);
        //3、这里的日志格式经过变换，形成（username，date username)的格式
        JavaPairDStream<String, String> userAdsClickLogDStream = adsClickLogDStream.mapToPair(
                new PairFunction<String, String, String>() {
            @Override
            public Tuple2<String, String> call(String adsClickLog) throws Exception {
                return new Tuple2<String, String>(adsClickLog.split(" ")[1], adsClickLog);
            }
        });

        //4、后面就可以进行transform操作了，将每个batch的RDD与黑名单进行join、filter等其他操作
        JavaDStream<String> validAdsClickLogDStream = userAdsClickLogDStream.transform(
                new Function<JavaPairRDD<String, String>, JavaRDD<String>>() {
            @Override
            public JavaRDD<String> call(JavaPairRDD<String, String> userAdsClickLogRDD) throws Exception {
                // 这里为什么用左外连接？
                // 因为，并不是每个用户都存在于黑名单中的
                // 所以，如果直接用join，那么没有存在于黑名单中的数据，会无法join到
                // 就给丢弃掉了
                // 所以，这里用leftOuterJoin，就是说，哪怕一个user不在黑名单RDD中，没有join到
                // 也还是会被保存下来的
                JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> joinedRDD =
                        userAdsClickLogRDD.leftOuterJoin(blacklistRDD);

                //连接之后，执行filter算子
                JavaPairRDD<String, Tuple2<String, Optional<Boolean>>> filteredRDD = joinedRDD.filter(
                                new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, Boolean>() {
                    @Override
                    public Boolean call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception {
                        if (tuple._2._2().isPresent() && tuple._2._2().get()) {
                            return false;
                        }
                        return true;
                    }
                });
                //过滤后的filteredRDD剩下的没有被黑名单的用户过滤了
                //后面进行map操作，转换成我们想要的格式
                JavaRDD<String> validAdsClickLogRDD = filteredRDD.map(new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, String>() {
                    @Override
                    public String call(Tuple2<String, Tuple2<String, Optional<Boolean>>> tuple) throws Exception {
                        return tuple._2._1;
                    }
                });
                return validAdsClickLogRDD;
            }
        });

        // 打印有效的广告点击日志
        // 其实在真实企业场景中，这里后面就可以走写入kafka、ActiveMQ等这种中间件消息队列
        // 然后再开发一个专门的后台服务，作为广告计费服务，执行实时的广告计费，这里就是只拿到了有效的广告点击
        validAdsClickLogDStream.print();

        jssc.start();
        jssc.awaitTermination();
        jssc.close();
    }
}
